package com.example.rabbitmq.demo2_deadQueue;

import com.example.rabbitmq.enums.ExchangeEnum;
import com.example.rabbitmq.enums.RabbitmqQueueNamesEnum;
import com.example.rabbitmq.enums.RoutingKeyEnum;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @Author songyahui
 * @Date 2023/7/11 11:51
 * @Description 模拟消息转到死信队列的情况，由Consumer2消费死信队列的消息
 */
public class Consumer1 {

    public static void consumeMessageWithDirect(Channel channel) throws IOException, TimeoutException {
        //声明死信和普通交换机 类型未direct
        channel.exchangeDeclare(ExchangeEnum.NORMAL_EXCHANGE.name, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(ExchangeEnum.DEAD_EXCHANGE.name, BuiltinExchangeType.DIRECT);
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String msg = new String(message.getBody(), "UTF-8");
            if(msg.equals("info-test-reject5")){
                System.out.println("普通队列-拒绝消息："+msg);
                // basic.nack方法为不确认deliveryTag对应的消息，第二个参数是否应用于多消息，第三个参数是否requeue.
                // 与basic.reject区别就是同时支持多个消息，可以nack该消费者先前接收未ack的所有消息。nack后的消息也会被自己消费到。
                // channel.basicNack(deliveryTag, false, true);
                // basic.reject方法拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列.
                // 该方法reject后，该消费者还是会消费到该条被reject的消息。
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            }else {
                System.out.println("普通消费："+msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }

        };
        //声明普通队列
        Map<String, Object> arguments = new HashMap<>();
        //过期时间
        //设置过期之后的死信交换机
        arguments.put("x-dead-letter-exchange", ExchangeEnum.DEAD_EXCHANGE.name);
        //设置死信routing key
        arguments.put("x-dead-letter-routing-key", RoutingKeyEnum.DEAD.name());
        //设置ttl，单位ms，也可以由生产方设置ttl，如果两边同时设置ttl则以最短时间为准
        arguments.put("x-message-ttl", 11100000);
        //设置消息长度限制
        arguments.put("x-max-length", 100);
        channel.queueDeclare(RabbitmqQueueNamesEnum.QUEUE_NORMAL.queueName, false, false, false, arguments);
        //声明死信队列
        channel.queueDeclare(RabbitmqQueueNamesEnum.QUEUE_DEAD.queueName, false, false, false, null);
        //绑定普通交换机和队列
        channel.queueBind(RabbitmqQueueNamesEnum.QUEUE_NORMAL.queueName, ExchangeEnum.NORMAL_EXCHANGE.name, RoutingKeyEnum.NORMAL.name());
        //绑定死信交换机和队列
        channel.queueBind(RabbitmqQueueNamesEnum.QUEUE_DEAD.queueName, ExchangeEnum.DEAD_EXCHANGE.name, RoutingKeyEnum.DEAD.name());
        //绑定死信交换机和队列
        channel.basicConsume(RabbitmqQueueNamesEnum.QUEUE_NORMAL.queueName, false, deliverCallback, consumerTag -> {});
//        channel.close();
    }
}
